/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.example.quickstart;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.Scanner;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * This class demonstrates how to send messages to brokers using provided {@link DefaultMQProducer}.
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException, IOException {

        /*
         * Instantiate with a producer group name.
         */
        DefaultMQProducer producer = new DefaultMQProducer("producerp");

        /*
         * Specify name server addresses.
         * <p/>
         *
         * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
         * <pre>
         * {@code
         * producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
         * }
         * </pre>
         */
         producer.setNamesrvAddr("localhost:9876");

        /*
         * Launch the instance.
         */
        producer.start();

        Scanner sc = new Scanner(System.in);
        AtomicInteger n = new AtomicInteger();
        while (true){
            String line = sc.nextLine();
            System.out.println("收到:"+line);
            if(line.equals("exit")){
                break;
            }

            String[] split = line.split(",");
            if(split.length!=3){
                continue;
            }

            for (int t = 0; t < Integer.parseInt(split[1]); t++) {
                new Thread(()->{
                    /*
                     * Call send message to deliver message to one of brokers.
                     */
                    try {
                        for (int i = 0; i < Integer.parseInt(split[2]); i++) {
                            try {
                                /*
                                 * Create a message instance, specifying topic, tag and message body.
                                 */
                                Message msg = new Message("t2" /* Topic */,
                                        split[0] /* Tag */,
                                        ("t2"+split[0]+"--"+n.incrementAndGet()+"--"+ DateFormatUtils.format(new Date(),"yyyy-MM-dd HH:mm:ss")).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                                );

                                SendResult sendResult = producer.send(msg);
                System.out.println("发送:"+sendResult);
                            } catch (MQClientException e) {
                                e.printStackTrace();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            } catch (UnsupportedEncodingException e) {
                                e.printStackTrace();
                            }
//                    System.out.println(sendResult);
                        }
                    } catch (RemotingException e) {
                        e.printStackTrace();
                    } catch (MQBrokerException e) {
                        e.printStackTrace();
                    }

                }).start();
            }


//                System.out.printf("%s%n", sendResult);

        }
        producer.shutdown();


        /*
         * Shut down once the producer instance is not longer in use.
         */
    }
}
